package skewreduce.framework;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.Writable;

public class MuxData implements Writable {
    DataInputBuffer inBuf = new DataInputBuffer();
    DataOutputBuffer outBuf = new DataOutputBuffer(8192);
    enum State {
        NOT_AVAILBLE,
        READY_KEY,
        READY_VALUE
    }
    State state;

    public <K extends Writable,V extends Writable> void set(K key,V val) throws IOException {
        outBuf.reset();
        key.write(outBuf);
        val.write(outBuf);
    }

    public void readFields(DataInput in) throws IOException {
        int sz = in.readInt();
        outBuf.reset();
        outBuf.write(in,sz);
        inBuf.reset(outBuf.getData(),sz);
        state = State.READY_KEY;
    }

    public void write(DataOutput out) throws IOException {
        out.writeInt(outBuf.getLength());
        out.write(outBuf.getData(),0,outBuf.getLength());
    }

    public void getKey(Writable key) throws IOException {
        if ( state != State.READY_KEY ) throw new IllegalStateException();
        key.readFields(inBuf);
        state = State.READY_VALUE;
    }

    public void getValue(Writable value) throws IOException {
        if ( state != State.READY_VALUE ) throw new IllegalStateException();
        value.readFields(inBuf);
        state = State.NOT_AVAILBLE;
    }

    public DataInput getDataInput() { return inBuf; }
}
